1 import gymnasium as gym
2 import numpy as np
3 import scipy
4 import tensorflow as tf
5 from tensorflow.keras.layers import Dense
6 import matplotlib.pyplot as plt
7
8 def neural_network_actor_critic(): # actor-critic 공유 신경망
9 input_x=tf.keras.Input(shape=(s_dim,))
10 hidden_x=Dense(32,activation='tanh')(input_x)
11 actor_y=Dense(a_dim,activation='softmax')(hidden_x)
12 critic_y=Dense(1,activation='linear')(hidden_x)
13 mlp=tf.keras.Model(inputs=input_x,outputs=[actor_y,critic_y])
14 return mlp
15
16 def discount_cumulation(x,discount):
17 return scipy.signal.lfilter([1],[1,-discount],x[::-1],axis=0)[::-1]
18
19 def advantage_MC(rewards,values,gamma): # 몬테카를로 이익 계산
20 return discount_cumulation(rewards,gamma)-np.array(values)
21
22 actionprob_mem=[] # 데이터 저장소
23 value_mem=[]
24 reward_mem=[]
25
26 def episode_generate_and_learn():
27 s,info=env.reset()
28 with tf.GradientTape (persistent=True) as tape:
29 while True:
30 action_prob,value=ac_model(s.reshape([1,s_dim]))
31 a=np.random.choice(a_dim,p=np.array(action_prob[0]))
32 s1,r,terminated,truncated,info=env.step(a)
33
34 actionprob_mem.append(action_prob[0,a])
35 value_mem.append(value[0][0])
36 reward_mem.append(r)
37 s=s1
38
39 if terminated or truncated:
40 break
41
42 returns=discount_cumulation(reward_mem,gamma)
43 advantage=advantage_MC(reward_mem,value_mem,gamma)
44 advantage=tf.convert_to_tensor(advantage,dtype=tf.float32)
45 returns=tf.convert_to_tensor(returns,dtype=tf.float32)
46
47 loss_actor=-advantage*tf.math.log(actionprob_mem)
48 loss_critic=tf.math.square(returns-value_mem)
49 loss=loss_actor+loss_critic
50
51 grad=tape.gradient(tf.reduce_mean(loss),ac_model.trainable_weights)
52 optimizer.apply_gradients(zip(grad,ac_model.trainable_weights))
53
54 gamma=0.99 # 할인율
55 n_episode=1000
56
57 env=gym.make('CartPole-v1')
58 s_dim=env.observation_space.shape[0] # 상태 공간
59 a_dim=env.action_space.n # 행동 공간
60
61 ac_model=neural_network_actor_critic() # actor-critic 신경망 생성
62 optimizer=tf.optimizers.Adam(learning_rate=0.01)
63
64 epi_length=[] # 에피소드 길이
65 for i in range(n_episode):
66 episode_generate_and_learn()
67 epi_length.append(len(actionprob_mem))
68 actionprob_mem,reward_mem,value_mem=[],[],[] # 데이터 비우기
69
70 if (i+1)%20==0: print(i+1,'번째 에피소드 평균 길이(최근 20개):',np.mean(epi_length[-20:]))
71 if np.min(epi_length[-5:])>=env.spec.max_episode_steps: # 연속 5번 최대 길이 넘으면 조기종료
72 break
73
74 ac_model.save('f9-3.keras')
75 env.close()
76
77 plt.figure(figsize=(16,5))
78 plt.plot(range(1,len(epi_length)+1),epi_length)
79 smooth=np.convolve(epi_length,10*[0.1],mode='valid')
80 plt.plot(range(1,len(smooth)+1),smooth)
81 plt.title('A2C scores for CartPole-v1 using shared network')
82 plt.ylabel('Score')
83 plt.xlabel('Episode')
84 plt.grid()
85 plt.show()
